Python Watchdog

2024-07-15 18:00| 来源: 网络整理| 查看: 265

文章目录简介安装初试重定向到日志中Handler类型命令行工具动态日志监控遇到的坑参考文献

简介

Watchdog是一款用于监控文件系统事件的Python库，对不同平台的事件进行了封装。

Watchdog优先使用底层原生API，其次再通过轮询磁盘实现监控，目前支持以下操作系统：

Windows：ReadDirectoryChangesWLinux 2.6+：inotifyMac OS X：kqueue 或 FSEventsBSD Unix variants：kqueue

仅支持Python 3.5+

安装 pip install watchdog

初试 import time from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class MyHandler(FileSystemEventHandler): def on_any_event(self, event): print(event.event_type, event.src_path) event_handler = MyHandler() observer = Observer() observer.schedule(event_handler, path='.', recursive=False) observer.start() try: while True: time.sleep(0.1) except KeyboardInterrupt: observer.stop() observer.join()

运行后，执行以下操作

新建文本文档改名为 a.txt输入内容 Hello World! 并保存删除 a.txt

效果：

created .\a.txt modified .\a.txt~ modified .\a.txt moved .\a.txt deleted .\b.txt

在这里插入图片描述

退出：Ctrl+C

PS：

PyCharm直接操作可能有延迟

重定向到日志中

输出重定向为logging日志中

import sys import time import logging from watchdog.observers import Observer from watchdog.events import LoggingEventHandler logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S') path = sys.argv[1] if len(sys.argv) > 1 else '.' event_handler = LoggingEventHandler() observer = Observer() observer.schedule(event_handler, path, recursive=True) observer.start() try: while True: time.sleep(0.1) finally: observer.stop() observer.join()

测试

import os import time n = 0 src = 'a.txt' dst = 'b.txt' if os.path.exists(src): os.remove(src) if os.path.exists(dst): os.remove(dst) while True: n += 1 with open(src, 'w') as f: print(f.write('Hello World!\n')) time.sleep(0.5) if n == 2: os.rename(src, dst) if n == 5: os.remove(src) os.remove(dst) n = 0

输出

2020-12-02 15:52:43 - Created file: .\a.txt 2020-12-02 15:52:43 - Modified file: .\a.txt 2020-12-02 15:52:44 - Modified file: .\a.txt 2020-12-02 15:52:44 - Modified file: .\a.txt 2020-12-02 15:52:45 - Modified file: .\a.txt 2020-12-02 15:52:45 - Modified file: .\a.txt 2020-12-02 15:52:46 - Deleted file: .\a.txt 2020-12-02 15:52:46 - Deleted file: .\b.txt

Handler类型类型功能FileSystemEventHandler文件PatternMatchingEventHandler模式匹配文件RegexMatchingEventHandler正则匹配文件LoggingEventHandler记录日志

命令行工具

安装

pip install watchdog[watchmedo]

查看命令

watchmedo --help

动态日志监控

思路：

打开文件将流位置改到末尾并记录当前位置watchdog监控一旦触发文件修改，将流位置改到末尾，记录当前位置根据步骤1和步骤3的位置获取新增片段解析新增片段

推荐阅读

Python一行代码实现Linux的taillogPatrol GitHub

不断写时间

import time import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s', datefmt='%H:%M:%S') logger = logging.getLogger() logger.setLevel(logging.INFO) fileHandler = logging.FileHandler('test.log', mode='a+') formatter = logging.Formatter('%(asctime)s', datefmt='%H:%M:%S') fileHandler.setFormatter(formatter) logger.addHandler(fileHandler) while True: logger.info('') time.sleep(1)

监控变化并输出内容

import os import time from watchdog.observers import Observer from watchdog.events import PatternMatchingEventHandler f = open('test.log', 'rb') f.seek(0, os.SEEK_END) # 将流位置改为末尾 last_position = f.tell() # 上次的流末尾位置 f.close() class MyHandler(PatternMatchingEventHandler): def __init__(self, patterns=['*.log'], ignore_patterns=None, ignore_directories=False, case_sensitive=False): super().__init__(patterns, ignore_patterns, ignore_directories, case_sensitive) def on_modified(self, event): global last_position f = open(event.src_path, 'rb') f.seek(0, os.SEEK_END) now_position = f.tell() if now_position == last_position: return read_bytes = now_position - last_position # 新增内容 f.seek(last_position, os.SEEK_SET) # 光标从开头移动 data = f.read(read_bytes).decode('utf-8') last_position = now_position f.close() print(data) if __name__ == '__main__': print('正在监控') event_handler = MyHandler() observer = Observer() observer.schedule(event_handler, path='.', recursive=False) observer.start() try: while True: time.sleep(0.1) except KeyboardInterrupt: observer.stop() observer.join()